package org.apache.kafka.niezhipeng.customSerializerDemo.serializer;

import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.niezhipeng.customSerializerDemo.pojo.Company;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;

/**
 * @author Nie ZhiPeng
 * @date 2025/7/25 0:06
 */
public class CompanySerializer implements Serializer<Company> {
    @Override
    public byte[] serialize(String topic, Company data) {
        if (data == null) {
            return null;
        }
        String name = data.getName();
        String address = data.getAddress();

        byte[] nameBytes = name == null ? new byte[0] : name.getBytes(StandardCharsets.UTF_8);
        byte[] addressBytes = address == null ? new byte[0] : address.getBytes(StandardCharsets.UTF_8);
        ByteBuffer byteBuffer = ByteBuffer.allocate(nameBytes.length + addressBytes.length + 4 + 4);
        byteBuffer.putInt(nameBytes.length);
        byteBuffer.put(nameBytes);
        byteBuffer.putInt(addressBytes.length);
        byteBuffer.put(addressBytes);
        return byteBuffer.array();
    }
}
